package com.bw.wjw.job;

import com.bw.wjw.function.AdjustIsNewProcessFunction;
import com.alibaba.fastjson.JSON;

import com.bw.wjw.KafkaUtil;
import com.bw.wjw.function.LogSplitProcessFunction;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;
public class JtpLogEtlJob {
    public static void main(String[] args) throws Exception {
//        1.执行环境-env
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
//        2.数据源-source
        DataStream<String> kafkaDataStream = KafkaUtil.consumerKafka(env, "topic-log");
        kafkaDataStream.print();
//        3.数据转换-transformation

//        4.数据接收器-sink
        processLog(kafkaDataStream) ;
//        5.执行环境-env
        env.execute("JtpLogEtlJob");
    }
/**
 * 对实时获取APP流量日志数据进行ETL处理，并且进行分流，存储Kafka消息队列
 *      1-数据清洗
 *      2-新老访客状态标记修复
 *      3-数据分流
 *      4-存储数据
 * @param logStream app日志流
 */
private static void processLog(DataStream<String> logStream) {
    // 1-数据清洗
    DataStream<String> jsonStream = logCleaned(logStream);

    // 2-新老访客状态标记修复
    DataStream<String> etlStream = processIsNew(jsonStream);
    // etlStream.print();

    // 3-数据分流
    DataStream<String> pageStream = splitStream(etlStream);
    // pageStream.print();

    // 4-存储数据
    KafkaUtil.producerKafka(pageStream, "dwd-log-page-view-log");
    }
    /**
     * 3-数据分流
     */
    private static DataStream<String> splitStream(DataStream<String> logStream) {
        // 侧流标签
        OutputTag<String> errorTag = new OutputTag<String>("error-tag"){};
        OutputTag<String> startTag = new OutputTag<String>("start-tag"){};
        OutputTag<String> displayTag = new OutputTag<String>("display-tag"){};
        OutputTag<String> actionTag = new OutputTag<String>("action-tag"){};

        // 日志分流
        SingleOutputStreamOperator<String> pageStream = logStream.process(
                new LogSplitProcessFunction(errorTag, startTag, displayTag, actionTag)
        );

        // 侧流输出
        DataStream<String> errorStream = pageStream.getSideOutput(errorTag);
        KafkaUtil.producerKafka(errorStream, "dwd-log-error-log");
        DataStream<String> startTStream = pageStream.getSideOutput(startTag);
        KafkaUtil.producerKafka(startTStream, "dwd-log-startT-log");
        DataStream<String> displayStream = pageStream.getSideOutput(displayTag);
        KafkaUtil.producerKafka(displayStream, "dwd-log-display-log");
        DataStream<String> actionStream = pageStream.getSideOutput(actionTag);
        KafkaUtil.producerKafka(actionStream, "dwd-log-action-log");

        // 返回主流
        return pageStream ;
    }
    /**
     * 2-新老访客状态标记修复
     */
    private static DataStream<String> processIsNew(DataStream<String> jsonStream) {
        // 首先，对数据量按照mid设备标识符进行分组
        KeyedStream<String, String> midStream = jsonStream.keyBy(
                value -> JSON.parseObject(value).getJSONObject("common").getString("mid")
        );

        // 然后，状态编程进行修复
        DataStream<String> isNewStream = midStream.process(new AdjustIsNewProcessFunction());

        // 最后，返回流
        return isNewStream ;
    }
    /**
     * 1-数据清洗
     */
    private static DataStream<String> logCleaned(DataStream<String> logStream) {
        // 侧边输出标记
        OutputTag<String> dirtyTag = new OutputTag<String>("dirty-tag"){} ;
        SingleOutputStreamOperator<String> jsonStream = logStream.process(
                new ProcessFunction<String, String>() {
                    @Override
                    public void processElement(String value,
                                               ProcessFunction<String, String>.Context ctx,
                                               Collector<String> out) throws Exception {
                        // value 表示流中每条数据；ctx 侧边流输出使用（上下文对象）；out 正常流数据输出使用；
                        try {
                            JSON.parseObject(value);
                            // 正常数据输出
                            out.collect(value);
                        } catch (Exception e) {
                            // 异常数据，侧边输出
                            ctx.output(dirtyTag, value);
                        }
                    }
                }
        );

        // 获取异常数据流
        DataStream<String> dirtyStream = jsonStream.getSideOutput(dirtyTag);
        KafkaUtil.producerKafka(dirtyStream, "dwd-log-dirty-log");

        //  返回数据流
        return jsonStream ;
    }


}